package com.atguigu.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Author: Pepsi
 * Date: 2023/8/24
 * Desc:
 */
public class Flink06_Table_Kafka {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        Schema schema = new Schema()
                .field("id", DataTypes.STRING())
                .field("ts", DataTypes.BIGINT())
                .field("vc", DataTypes.INT());

        tEnv
                .connect(
                        new Kafka()
                                .version("universal")
                        .property("bootstrap.servers","hadoop101:9092")
                        .property("group.id","flink_consumer_group")
                        .topic("flink_source_kafka")
                        .startFromLatest()
                )
//                .withFormat(new Csv())
                // 读取json格式的
                .withFormat(new Json())
                .withSchema(schema)
                .createTemporaryTable("sensor");

        Table table = tEnv.from("sensor").where($("vc").isGreater(10)).select("*");
//        table.execute().print();


        tEnv
                .connect(
                        new Kafka()
                        .version("universal")
                        .property("bootstrap.servers","hadoop101:9092")
                        .topic("flink_sink_kafka")
                )
                .withFormat(new Json())
                .withSchema(schema)
                .createTemporaryTable("abc");

        table.executeInsert("abc");
    }
}
